﻿/***************************
 * Author：peiyang
 * DateTime：
 * Description：<Description>
 * Copyright：臭皮匠工作室@2020
 **/

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Framing;
using Cpjit.SunnyPaine.Toolkit.RabbitMQ.Enums;
using Cpjit.SunnyPaine.Toolkit.RabbitMQ.Exceptions;
using Cpjit.SunnyPaine.Toolkit.RabbitMQ.Model;

namespace Cpjit.SunnyPaine.Toolkit.RabbitMQ.Impl
{
    /// <summary>
    /// RabbitMQ客户端。
    /// </summary>
    public class RabbitmqClient : IRabbitmqClient
    {
        #region 私有变量
        /// <summary>
        /// 是否运行。
        /// </summary>
        private bool isRunning = false;

        /// <summary>
        /// 连接工厂。
        /// </summary>
        private ConnectionFactory factory;

        /// <summary>
        /// 连接对象。
        /// </summary>
        private IConnection connection;

        /// <summary>
        /// 表示由多个线程同时访问的 work 消息模式的生产者键值对的线程安全集合。
        /// <para>键：队列名称。值：生产者消息通道模型。</para>
        /// </summary>
        private readonly ConcurrentDictionary<string, ChannelMetaObject> producerWorkDict;

        /// <summary>
        /// 表示由多个线程同时访问的 fanout 消息模式的生产者键值对的线程安全集合。
        /// <para>键：交换器名称。值：生产者消息通道模型。</para>
        /// </summary>
        private readonly ConcurrentDictionary<string, ChannelMetaObject> producerFanoutDict;

        /// <summary>
        /// 表示由多个线程同时访问的 direct 消息模式的生产者键值对对的线程安全集合。
        /// <para>键：交换器名称。值：生产者消息通道模型。</para>
        /// </summary>
        private readonly ConcurrentDictionary<string, ChannelMetaObject> producerDirectDict;

        /// <summary>
        /// 表示由多个线程同时访问的 topic 消息模式的生产者键值对的线程安全集合。
        /// <para>键：交换器名称。值：生产者消息通道模型。</para>
        /// </summary>
        private readonly ConcurrentDictionary<string, ChannelMetaObject> producerTopicDict;

        /// <summary>
        /// 表示由多个线程同时访问的 work 消息模式的消费者键值对的线程安全集合。
        /// <para>键：队列名称。值：消费者模型。</para>
        /// </summary>
        private readonly ConcurrentDictionary<string, IConsumer> consumerWorkDict;

        /// <summary>
        /// 表示由多个线程同时访问的 fanout 消息模式的消费者键值对的线程安全集合。
        /// <para>键：交换器名称。值：消费者消息通道模型。</para>
        /// </summary>
        private readonly ConcurrentDictionary<string, IConsumer> consumerFanoutDict;

        /// <summary>
        /// 表示由多个线程同时访问的 direct 消息模式的消费者键值对的线程安全集合。
        /// <para>键：交换器名称。值：消费者消息通道模型。</para>
        /// </summary>
        private readonly ConcurrentDictionary<string, IConsumer> consumerDirectDict;

        /// <summary>
        /// 表示由多个线程同时访问的 topic 消息模式的消费者键值对的线程安全集合。
        /// <para>键：交换器名称。值：消费者消息通道模型。</para>
        /// </summary>
        private readonly ConcurrentDictionary<string, IConsumer> consumerTopicDict;
        #endregion


        #region 公共属性，IRabbitmqClient成员
        /// <summary>
        /// RabbitMQ的主机名或远程IP地址。
        /// </summary>
        public string HostName { get; set; }

        /// <summary>
        /// 端口号
        /// </summary>
        public int Port { get; set; }

        /// <summary>
        /// 用户名。
        /// </summary>
        public string UserName { get; set; }

        /// <summary>
        /// 密码。
        /// </summary>
        public string Password { get; set; }

        /// <summary>
        /// 客户端唯一标识。
        /// </summary>
        public string ClientId { get; set; }

        /// <summary>
        /// 消息编码。默认 System.Text.Encoding.Default 。
        /// </summary>
        public Encoding Encoding { get; set; }

        /// <summary>
        /// 心跳检测周期。单位：秒。
        /// </summary>
        public ushort? RequestedHeartbeat { get; set; }

        /// <summary>
        /// 是否处于已连接可用状态。
        /// </summary>
        public bool IsConnected { get; private set; }

        /// <summary>
        /// 初始连接时，如果连接失败，执行重连时的重连周期。默认2000ms。单位ms。
        /// </summary>
        public int ConnectRetryInterval { get; set; } = 2000;
        #endregion


        #region 委托，IRabbitmqClient成员
        /// <summary>
        /// 当与RabbitMQ连接状态变化时发生。
        /// </summary>
        public event Action<IRabbitmqClient, StatusEventArgs> ConnectionStatusChanged;

        /// <summary>
        /// 当出现错误时发生。
        /// <para>IRabbitmqClient：RabbitMQ客户端对象。ExceptionEventArgs：异常信息。bool：是否可用于UI操作。</para>
        /// </summary>
        public event Action<IRabbitmqClient, ExceptionEventArgs, bool> Error;
        #endregion


        #region 构造方法
        /// <summary>
        /// 使用默认参数创建实例。
        /// </summary>
        public RabbitmqClient() : this("127.0.0.1", 5672)
        {
        }

        /// <summary>
        /// 使用指定的参数创建实例。
        /// </summary>
        /// <param name="hostName">主机名称，或者IP。</param>
        /// <param name="port">端口。</param>
        public RabbitmqClient(string hostName, int port)
            : this(hostName, port, null, null)
        {
            this.HostName = hostName;
            this.Port = port;
        }

        /// <summary>
        /// 使用指定的参数创建实例
        /// </summary>
        /// <param name="hostName">主机名称，或者IP。</param>
        /// <param name="port">端口。</param>
        /// <param name="userName">用户名。</param>
        /// <param name="password">密码。</param>
        public RabbitmqClient(string hostName, int port, string userName, string password)
            : this(hostName, port, userName, password, null)
        {
        }

        /// <summary>
        /// 使用指定的参数与创建实例。
        /// </summary>
        /// <param name="hostName">主机名称，或者IP。</param>
        /// <param name="port">端口。</param>
        /// <param name="userName">用户名。</param>
        /// <param name="password">密码。</param>
        /// <param name="clientId">客户端Id。</param>
        public RabbitmqClient(string hostName, int port, string userName, string password, string clientId)
        {
            this.HostName = hostName;
            this.Port = port;
            this.UserName = userName;
            this.Password = password;
            this.ClientId = clientId;

            this.Encoding = Encoding.Default;

            this.producerWorkDict = new ConcurrentDictionary<string, ChannelMetaObject>();
            this.producerFanoutDict = new ConcurrentDictionary<string, ChannelMetaObject>();
            this.producerDirectDict = new ConcurrentDictionary<string, ChannelMetaObject>();
            this.producerTopicDict = new ConcurrentDictionary<string, ChannelMetaObject>();

            this.consumerWorkDict = new ConcurrentDictionary<string, IConsumer>();
            this.consumerFanoutDict = new ConcurrentDictionary<string, IConsumer>();
            this.consumerDirectDict = new ConcurrentDictionary<string, IConsumer>();
            this.consumerTopicDict = new ConcurrentDictionary<string, IConsumer>();
        }
        #endregion


        #region 析构方法
        /// <summary>
        /// 析构方法。释放对象。
        /// </summary>
        ~RabbitmqClient()
        {
            this.isRunning = false;
        }
        #endregion


        #region 委托方法
        /// <summary>
        /// 当连接恢复时发生。
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="args"></param>
        private void Connection_RecoverySucceeded(object sender, EventArgs args)
        {
            this.IsConnected = true;
            this.ConnectionStatusChanged?.Invoke(this, new StatusEventArgs(true));
        }

        /// <summary>
        /// 当连接关闭/断开时发生。
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="args"></param>
        private void Connection_ConnectionShutdown(object sender, ShutdownEventArgs args)
        {
            this.IsConnected = false;
            this.ConnectionStatusChanged?.Invoke(this, new StatusEventArgs(false));
        }
        #endregion


        #region 私有方法
        /// <summary>
        /// 创建通道模型。
        /// </summary>
        /// <param name="workMode">消息工作模式。</param>
        /// <param name="exchangeName">交换器名称。如果消息工作模式基于交换器的话。</param>
        /// <param name="queueName">队列名称。</param>
        /// <param name="routingKey">路由键，如果消息工作模式基于交换器的定向路由模式（direct）的话。通配符，如果消息工作模式基于交换器的通配符模式（topic）的话。</param>
        /// <param name="arguments">附加信息。</param>
        /// <returns></returns>
        /// <exception cref="ArgumentNullException">当将 null 引用传递到不接受其作为有效参数的方法时引发的异常。</exception>
        /// <exception cref="NotSupportedException">当 WokrMode 类型不受不支持所调用功能时引发的异常。</exception>
        /// <exception cref="ArgumentOutOfRangeException">当 WorkMode 类型的值超出了所调用方法定义的允许值范围时引发的异常。</exception>
        private ChannelMetaObject CreateChanel(WorkMode workMode, string exchangeName, string queueName, string routingKey,
            IDictionary<string, object> arguments)
        {
            ChannelMetaObject metaObject;
            switch (workMode)
            {
                case WorkMode.Work:
                    {
                        if (this.producerWorkDict.ContainsKey(queueName))
                        {
                            metaObject = this.producerWorkDict[queueName];
                        }
                        else //如果指定目标名称的生产者在 work 生产者键值表中不存在
                        {
                            if (string.IsNullOrEmpty(queueName))
                            {
                                throw new ArgumentNullException(queueName);
                            }

                            IModel channel = this.connection.CreateModel();
                            channel.QueueDeclare(queueName, false, false, true, arguments);

                            metaObject = new ChannelMetaObject
                            {
                                Arguments = arguments,
                                Channel = channel,
                                Mode = workMode,
                                QueueName = queueName
                            };
                            this.producerWorkDict.TryAdd(queueName, metaObject);
                        }
                    }
                    break;
                case WorkMode.Fanout:
                    {
                        if (this.producerFanoutDict.ContainsKey(exchangeName))
                        {
                            metaObject = this.producerFanoutDict[exchangeName];
                            if (!string.IsNullOrEmpty(queueName) && metaObject.QueueNames.Value.IndexOf(queueName) < 0) //如果当前通道模型的交换器没有声明该队列
                            {
                                metaObject.Channel.QueueDeclare(queueName, false, false, true, arguments);
                                metaObject.Channel.QueueBind(queueName, exchangeName, "");
                                metaObject.QueueNames.Value.Add(queueName);
                            }
                        }
                        else //如果指定目标名称的生产者在 fanout 生产者键值表中不存在
                        {
                            if (string.IsNullOrEmpty(exchangeName))
                            {
                                throw new ArgumentNullException(nameof(exchangeName));
                            }

                            IModel channel = this.connection.CreateModel();
                            channel.ExchangeDeclare(exchangeName, workMode.ToString().ToLower(), false, false, arguments);
                            if (!string.IsNullOrEmpty(queueName))
                            {
                                channel.QueueDeclare(queueName, false, false, true, arguments);
                                channel.QueueBind(queueName, exchangeName, "");
                            }

                            metaObject = new ChannelMetaObject
                            {
                                Arguments = arguments,
                                Channel = channel,
                                Mode = workMode,
                                QueueNames = new KeyValuePair<string, List<string>>(exchangeName, new List<string>() { queueName })
                            };
                            this.producerFanoutDict.TryAdd(exchangeName, metaObject);
                        }
                    }
                    break;
                case WorkMode.Direct:
                    if (this.producerDirectDict.ContainsKey(exchangeName))
                    {
                        metaObject = this.producerDirectDict[exchangeName];
                        if (!string.IsNullOrEmpty(queueName) && metaObject.QueueNames.Value.IndexOf(queueName) < 0) //如果当前通道模型的交换器没有声明该队列
                        {
                            if (string.IsNullOrEmpty(routingKey))
                            {
                                throw new ArgumentNullException(nameof(routingKey));
                            }
                            metaObject.Channel.QueueDeclare(queueName, false, false, true, arguments);
                            metaObject.Channel.QueueBind(queueName, exchangeName, routingKey);
                            metaObject.QueueNames.Value.Add(queueName);
                        }
                    }
                    else //如果指定目标名称的生产者在 direct 生产者键值表中不存在
                    {
                        if (string.IsNullOrEmpty(exchangeName))
                        {
                            throw new ArgumentNullException(nameof(exchangeName));
                        }
                        if (string.IsNullOrEmpty(routingKey))
                        {
                            throw new ArgumentNullException(nameof(routingKey));
                        }

                        IModel channel = this.connection.CreateModel();
                        channel.ExchangeDeclare(exchangeName, workMode.ToString().ToLower(), false, false, arguments);
                        channel.QueueDeclare(queueName, false, false, true, arguments);
                        channel.QueueBind(queueName, exchangeName, routingKey);

                        metaObject = new ChannelMetaObject
                        {
                            Arguments = arguments,
                            Channel = channel,
                            Mode = workMode,
                            QueueNames = new KeyValuePair<string, List<string>>(exchangeName, new List<string>() { queueName })
                        };
                        this.producerDirectDict.TryAdd(exchangeName, metaObject);
                    }
                    break;
                case WorkMode.Topic:
                    {
                        if (this.producerTopicDict.ContainsKey(exchangeName))
                        {
                            metaObject = this.producerTopicDict[exchangeName];

                            if (string.IsNullOrEmpty(queueName))
                            {
                                throw new ArgumentNullException(nameof(queueName));
                            }
                            if (metaObject.QueueNames.Value.IndexOf(queueName) < 0)
                            {
                                if (string.IsNullOrEmpty(routingKey))
                                {
                                    throw new ArgumentNullException(nameof(routingKey));
                                }

                                metaObject.Channel.QueueDeclare(queueName, false, false, true, arguments);
                                metaObject.Channel.QueueBind(queueName, exchangeName, routingKey);
                                metaObject.QueueNames.Value.Add(queueName);
                            }
                        }
                        else //如果指定目标名称的生产者在 topic 生产者键值表中不存在
                        {
                            if (string.IsNullOrEmpty(exchangeName))
                            {
                                throw new ArgumentNullException(nameof(exchangeName));
                            }
                            if (string.IsNullOrEmpty(queueName))
                            {
                                throw new ArgumentNullException(nameof(queueName));
                            }
                            if (string.IsNullOrEmpty(routingKey))
                            {
                                throw new ArgumentNullException(nameof(routingKey));
                            }

                            IModel channel = this.connection.CreateModel();
                            channel.ExchangeDeclare(exchangeName, workMode.ToString().ToLower(), false, false, arguments);
                            channel.QueueDeclare(queueName, false, false, true, arguments);
                            channel.QueueBind(queueName, exchangeName, routingKey);

                            metaObject = new ChannelMetaObject
                            {
                                Arguments = arguments,
                                Channel = channel,
                                Mode = workMode,
                                QueueNames = new KeyValuePair<string, List<string>>(exchangeName, new List<string>() { queueName })
                            };
                            this.producerWorkDict.TryAdd(queueName, metaObject);
                        }
                    }
                    break;
                case WorkMode.RPC:
                    throw new NotSupportedException("当前版本不支持 RPC 工作模式。");
                default:
                    throw new ArgumentOutOfRangeException(nameof(workMode), workMode, null);
            }
            return metaObject;
        }
        #endregion


        #region 公共方法，IRabbitmqClient成员
        /// <summary>
        /// 连接RabbitMQ一次。
        /// </summary>
        /// <returns></returns>
        public bool ConnectOnce()
        {
            if (string.IsNullOrEmpty(this.HostName))
            {
                throw new ArgumentException("指向RabbitMQ服务的HostName不合法。HostName为主机名或者IP地址。");
            }
            if (this.Port < 1 || this.Port > 65535)
            {
                throw new ArgumentException("指向RabbitMQ服务的端口不合法。");
            }

            try
            {
                this.isRunning = true;
                this.factory = new ConnectionFactory()
                {
                    HostName = this.HostName,
                    Port = this.Port
                };
                if (this.UserName != null)
                {
                    this.factory.UserName = this.UserName;
                }
                if (this.Password != null)
                {
                    this.factory.Password = this.Password;
                }
                if (this.RequestedHeartbeat != null)
                {
                    this.factory.RequestedHeartbeat = this.RequestedHeartbeat.Value;
                }

                this.factory.RequestedConnectionTimeout = 2000;//连接超时2秒
                this.factory.AutomaticRecoveryEnabled = true;//是否自动重连
                this.connection = this.factory.CreateConnection(this.ClientId);
                this.connection.RecoverySucceeded += this.Connection_RecoverySucceeded;
                this.connection.ConnectionShutdown += this.Connection_ConnectionShutdown;
                this.IsConnected = true;
            }
            catch (Exception ex)
            {
                this.Error?.Invoke(this, new ExceptionEventArgs() { Exception = ex, Message = $"连接 {this.HostName}:{this.Port} 失败。" }, true);
            }
            return this.IsConnected;
        }

        /// <summary>
        /// 连接RabbitMQ。如果连接失败，会一直阻塞，直到重连成功。
        /// </summary>
        public bool Connect()
        {
            if (string.IsNullOrEmpty(this.HostName))
            {
                throw new ArgumentException("指向RabbitMQ服务的HostName不合法。HostName为主机名或者IP地址。");
            }
            if (this.Port < 1 || this.Port > 65535)
            {
                throw new ArgumentException("指向RabbitMQ服务的端口不合法。");
            }

            try
            {
                this.isRunning = true;
                this.factory = new ConnectionFactory()
                {
                    HostName = this.HostName,
                    Port = this.Port
                };
                if (this.UserName != null)
                {
                    this.factory.UserName = this.UserName;
                }
                if (this.Password != null)
                {
                    this.factory.Password = this.Password;
                }
                if (this.RequestedHeartbeat != null)
                {
                    this.factory.RequestedHeartbeat = this.RequestedHeartbeat.Value;
                }

                this.factory.RequestedConnectionTimeout = 2000;//连接超时2秒
                this.factory.AutomaticRecoveryEnabled = true;//是否自动重连
                this.connection = this.factory.CreateConnection(this.ClientId);
                this.connection.RecoverySucceeded += this.Connection_RecoverySucceeded;
                this.connection.ConnectionShutdown += this.Connection_ConnectionShutdown;
                this.IsConnected = true;
                return this.IsConnected;
            }
            catch (Exception ex)
            {
                this.Error?.Invoke(this,
                    new ExceptionEventArgs()
                    {
                        Exception = ex,
                        Message = $"连接 {this.HostName}:{this.Port} 失败。正在重试..."
                    },
                    true);
                Task task = Task.Factory.StartNew(() =>
                {
                    long retry = 1;
                    while (this.isRunning && this.IsConnected == false)
                    {
                        try
                        {
                            this.connection = this.factory.CreateConnection(this.ClientId);
                            this.connection.RecoverySucceeded += this.Connection_RecoverySucceeded;
                            this.connection.ConnectionShutdown += this.Connection_ConnectionShutdown;
                            this.IsConnected = true;
                        }
                        catch (Exception innerEx)
                        {
                            this.Error?.Invoke(this, new ExceptionEventArgs()
                            {
                                Exception = innerEx,
                                Message = $"重连 {this.HostName}:{this.Port} 失败（第{retry++}次）。",
                            }, false);
                        }
                        Thread.Sleep(this.ConnectRetryInterval);
                    }
                });
                task.Wait();
                return this.IsConnected;
            }
        }

        /// <summary>
        /// 异步连接。如果连接失败，会一直阻塞，直到重连成功。
        /// </summary>
        public void AsyncConnect(IEnumerable<string> dlls, bool autoRegisterConsumer = true)
        {
            Task.Factory.StartNew(() =>
            {
                this.Connect();
                if (autoRegisterConsumer)
                {
                    if (dlls != null)
                    {
                        this.RegisterConsumer(dlls);
                    }
                    else
                    {
                        this.RegisterConsumer();
                    }
                }
            });
        }

        /// <summary>
        /// 断开连接RabbitMQ。
        /// </summary>
        public void Close()
        {
            this.isRunning = false;
            this.IsConnected = false;

            //释放消费者
            if (this.consumerDirectDict != null && this.consumerDirectDict.Count > 0)
            {
                foreach (KeyValuePair<string, IConsumer> kv in this.consumerDirectDict)
                {
                    kv.Value.Dispose();
                }
            }
            if (this.consumerFanoutDict != null && this.consumerFanoutDict.Count > 0)
            {
                foreach (KeyValuePair<string, IConsumer> kv in this.consumerFanoutDict)
                {
                    kv.Value.Dispose();
                }
            }
            if (this.consumerTopicDict != null && this.consumerTopicDict.Count > 0)
            {
                foreach (KeyValuePair<string, IConsumer> kv in this.consumerTopicDict)
                {
                    kv.Value.Dispose();
                }
            }
            if (this.consumerWorkDict != null && this.consumerWorkDict.Count > 0)
            {
                foreach (KeyValuePair<string, IConsumer> kv in this.consumerWorkDict)
                {
                    kv.Value.Dispose();
                }
            }
            this.consumerDirectDict.Clear();
            this.consumerFanoutDict.Clear();
            this.consumerTopicDict.Clear();
            this.consumerWorkDict.Clear();

            //释放 work 通道
            if (this.producerWorkDict != null && this.producerWorkDict.Count > 0)
            {
                foreach (KeyValuePair<string, ChannelMetaObject> kv in this.producerWorkDict)
                {
                    if (kv.Value.Channel != null && kv.Value.Channel.IsOpen)
                    {
                        kv.Value.Channel.Close();
                        kv.Value.Channel.Dispose();
                    }
                }
            }
            this.producerWorkDict.Clear();
            //释放 direct 通道
            if (this.producerDirectDict != null && this.producerDirectDict.Count > 0)
            {
                foreach (KeyValuePair<string, ChannelMetaObject> kv in this.producerDirectDict)
                {
                    if (kv.Value.Channel != null && kv.Value.Channel.IsOpen)
                    {
                        kv.Value.Channel.Close();
                        kv.Value.Channel.Dispose();
                    }
                }
            }
            this.producerDirectDict.Clear();
            //释放 fanout 通道
            if (this.producerFanoutDict != null && this.producerFanoutDict.Count > 0)
            {
                foreach (KeyValuePair<string, ChannelMetaObject> kv in this.producerFanoutDict)
                {
                    if (kv.Value.Channel != null && kv.Value.Channel.IsOpen)
                    {
                        kv.Value.Channel.Close();
                        kv.Value.Channel.Dispose();
                    }
                }
            }
            this.producerFanoutDict.Clear();
            //释放 topic 通道
            if (this.producerTopicDict != null && this.producerTopicDict.Count > 0)
            {
                foreach (KeyValuePair<string, ChannelMetaObject> kv in this.producerTopicDict)
                {
                    if (kv.Value.Channel != null && kv.Value.Channel.IsOpen)
                    {
                        kv.Value.Channel.Close();
                        kv.Value.Channel.Dispose();
                    }
                }
            }
            this.producerTopicDict.Clear();

            if (this.connection != null && this.connection.IsOpen)
            {
                this.connection.Close();
                this.connection.Dispose();
            }
        }

        /// <summary>
        /// 注册消费者。如果使用IConsumer接口注册消费者，则必须调用该接口。
        /// </summary>
        public void RegisterConsumer()
        {
            IEnumerable<string> files = Directory.GetFiles(AppDomain.CurrentDomain.BaseDirectory);
            files = files.Where(p => p.EndsWith(".dll") || p.EndsWith(".exe"));
            this.RegisterConsumer(files);
        }

        /// <summary>
        /// 注册消费者。如果使用IConsumer接口注册消费者，则必须调用该接口。
        /// </summary>
        /// <param name="dlls">要扫描的目标dll完整路径。</param>
        public void RegisterConsumer(IEnumerable<string> dlls)
        {
            if (!this.IsConnected)
            {
                return;
            }
            List<Assembly> assemblies = new List<Assembly>();
            foreach (string dll in dlls)
            {
                assemblies.Add(Assembly.LoadFile(dll));
            }

            foreach (Assembly assembly in assemblies)
            {
                Type[] types = assembly.GetTypes()
                    .Where(p => !p.IsAbstract && p.IsPublic && typeof(IConsumer).IsAssignableFrom(p))
                    .ToArray();
                foreach (Type type in types)
                {
                    if (!typeof(IConsumer).IsAssignableFrom(type))
                    {
                        continue;
                    }
                    IConsumer consumer = (IConsumer)Activator.CreateInstance(type);
                    if (!consumer.IsSubscribed)
                    {
                        continue;
                    }
                    this.CreateConsumer(consumer);
                }
            }
        }

        /// <summary>
        /// 创建消费者。
        /// </summary>
        /// <param name="consumer">消费者对象。</param>
        public void CreateConsumer(IConsumer consumer)
        {
            ChannelMetaObject channelMetaObject = this.CreateChanel(consumer.Mode,
                        consumer.ExchangeName,
                        consumer.QueueName,
                        consumer.RoutingKey,
                        consumer.Arguments);
            ConsumerMetaObject consumerMetaObject = new ConsumerMetaObject
            {
                Channel = channelMetaObject.Channel,
                ExchangeName = consumer.ExchangeName,
                QueueName = consumer.QueueName,
                Arguments = consumer.Arguments,
                Mode = consumer.Mode,
                AutoAck = consumer.AutoAck
            };
            consumer.MetaObject = consumerMetaObject;
            consumer.RabbitmqClient = this;
            switch (consumer.Mode)
            {
                case WorkMode.Work:
                    if (string.IsNullOrEmpty(consumer.QueueName))
                    {
                        throw new CreateConsumerException("当消息模式为 work 时，消费者必须指明 QueueName 。");
                    }
                    if (this.consumerWorkDict.ContainsKey(consumer.QueueName))
                    {
                        throw new DuplicateNameException($"在 work 消息工作模式的消费者字典中，已经存在队列名为 {consumer.QueueName} 的消费者了。");
                    }
                    this.consumerWorkDict.TryAdd(consumer.QueueName, consumer);
                    break;
                case WorkMode.Fanout:
                    {
                        if (string.IsNullOrEmpty(consumer.ExchangeName))
                        {
                            throw new CreateConsumerException("当消息模式为 fanout 时，消费者必须指明 ExchangeName 。");
                        }
                        if (string.IsNullOrEmpty(consumer.QueueName))
                        {
                            throw new CreateConsumerException("当消息模式为 fanout 时，消费者必须指明 QueueName 。");
                        }
                        string key = $"{consumer.ExchangeName}__{consumer.QueueName}";
                        if (this.consumerFanoutDict.ContainsKey(key))
                        {
                            throw new DuplicateNameException($"在 fanout 消息工作模式的消费者字典中，已经存在 交换器名为 {consumer.ExchangeName}，" +
                                                             $"队列名为 {consumer.QueueName} 的消费者了。");
                        }
                        this.consumerFanoutDict.TryAdd(key, consumer);
                    }
                    break;
                case WorkMode.Direct:
                    {
                        if (string.IsNullOrEmpty(consumer.ExchangeName))
                        {
                            throw new CreateConsumerException("当消息模式为 direct 时，消费者必须指明 ExchangeName 。");
                        }
                        if (string.IsNullOrEmpty(consumer.QueueName))
                        {
                            throw new CreateConsumerException("当消息模式为 direct 时，消费者必须指明 QueueName 。");
                        }
                        string key = $"{consumer.ExchangeName}__{consumer.QueueName}";
                        if (this.consumerDirectDict.ContainsKey(key))
                        {
                            throw new DuplicateNameException($"在 direct 消息工作模式的消费者字典中，已经存在 交换器名为 {consumer.ExchangeName}，" +
                                                             $"队列名为 {consumer.QueueName} 的消费者了。");
                        }
                        this.consumerDirectDict.TryAdd(key, consumer);
                    }
                    break;
                case WorkMode.Topic:
                    {
                        if (string.IsNullOrEmpty(consumer.ExchangeName))
                        {
                            throw new CreateConsumerException("当消息模式为 topic 时，消费者必须指明 ExchangeName 。");
                        }
                        if (string.IsNullOrEmpty(consumer.QueueName))
                        {
                            throw new CreateConsumerException("当消息模式为 topic 时，消费者必须指明 QueueName 。");
                        }
                        string key = $"{consumer.ExchangeName}__{consumer.QueueName}";
                        if (this.consumerTopicDict.ContainsKey(key))
                        {
                            throw new DuplicateNameException($"在 topic 消息工作模式的消费者字典中，已经存在 交换器名为 {consumer.ExchangeName}，" +
                                                             $"队列名为 {consumer.QueueName} 的消费者了。");
                        }
                        this.consumerTopicDict.TryAdd(key, consumer);
                    }
                    break;
                case WorkMode.RPC:
                    throw new NotSupportedException("当前版本不支持 RPC 工作模式。");
                default:
                    throw new ArgumentOutOfRangeException();
            }
            this.BindDestination(consumerMetaObject, consumer.Receive);
        }

        /// <summary>
        /// 绑定目标。
        /// </summary>
        /// <param name="metaObject"></param>
        /// <param name="handler">目标方法。</param>
        public void BindDestination(ConsumerMetaObject metaObject, Action<IRabbitmqClient, DataEventArgs> handler)
        {
            EventingBasicConsumer consumer = new EventingBasicConsumer(metaObject.Channel);
            consumer.Received += (s, e) =>
            {
                if (handler == null)
                {
                    return;
                }
                DataEventArgs args = new DataEventArgs()
                {
                    Data = e.Body,
                    Content = this.Encoding.GetString(e.Body)
                };
                try
                {
                    handler(this, args);
                }
                catch (Exception ex)
                {
                    this.Error?.Invoke(this, new ExceptionEventArgs { Exception = ex, Message = "消费者接收到消息，处理消息过程中发生问题。" }, true);
                }
            };
            metaObject.Consumer = consumer;
            metaObject.Channel.BasicConsume(metaObject.QueueName, metaObject.AutoAck, consumer);
        }

        /// <summary>
        /// 发送文本消息。针对 simple/work 消息模式。
        /// </summary>
        /// <param name="text">文本消息。</param>
        /// <param name="queueName">队列名称。</param>
        /// <param name="property">附加属性。可为null。</param>
        public void Send(string text, string queueName, BasicProperty property)
        {
            this.Send(this.Encoding.GetBytes(text), WorkMode.Work, null, queueName, null, property);
        }

        /// <summary>
        /// 发送文本消息。针对基于交换器的消息模式。
        /// </summary>
        /// <param name="text">文本消息。</param>
        /// <param name="mode">工作模式。</param>
        /// <param name="exchangeName">交换器名称。如果工作模式为基于交换器的话。</param>
        /// <param name="routingKey">路由键。如果工作模式为基于交换器的 direct 模式的话。</param>
        /// <param name="property">附加属性。可为null。</param>
        public void Send(string text, WorkMode mode, string exchangeName, string routingKey, BasicProperty property)
        {
            if (mode == WorkMode.Work)
            {
                throw new ArgumentException("该方法不支持 work 消息模式。");
            }
            this.Send(this.Encoding.GetBytes(text), mode, exchangeName, null, routingKey, property);
        }

        /// <summary>
        /// 发送字节数组消息。
        /// </summary>
        /// <param name="data">消息数据。</param>
        /// <param name="mode">工作模式。</param>
        /// <param name="exchangeName">交换器名称。如果工作模式为基于交换器的话。</param>
        /// <param name="queueName">队列名称。</param>
        /// <param name="routingKey">路由键。如果工作模式为基于交换器的 direct 模式的话。</param>
        /// <param name="property">附加属性。可为null。</param>
        public void Send(byte[] data, WorkMode mode, string exchangeName, string queueName, string routingKey, BasicProperty property)
        {
            try
            {
                IBasicProperties properties = null;
                IDictionary<string, object> arguments = null;
                if (property != null)
                {
                    properties = new BasicProperties();
                    arguments = new Dictionary<string, object>();
                    //将property转换成arugments 和 Dictionary
                    if (!string.IsNullOrEmpty(property.UserId))
                    {
                        properties.UserId = property.UserId;
                        arguments["UserId"] = properties.UserId;
                    }
                    if (property.Timestamp != null)
                    {
                        properties.Timestamp = new AmqpTimestamp(property.Timestamp.Value);
                        arguments["Timestamp"] = properties.Timestamp;
                    }
                    if (!string.IsNullOrEmpty(property.ReplyTo))
                    {
                        properties.ReplyToAddress = new PublicationAddress(mode.ToString().ToLower(), exchangeName, routingKey);
                        properties.ReplyTo = property.ReplyTo;
                        arguments["ReplyToAddress"] = properties.ReplyTo;
                        arguments["ReplyTo"] = properties.ReplyTo;
                    }
                    if (property.Priotity != null)
                    {
                        properties.Priority = property.Priotity.Value;
                        arguments["Priority"] = properties.Priority;
                    }
                    if (property.Persistent != null)
                    {
                        properties.Persistent = property.Persistent.Value;
                        arguments["Persistent"] = properties.Persistent;
                    }
                    if (!string.IsNullOrEmpty(property.MessageId))
                    {
                        properties.MessageId = property.MessageId;
                        arguments["MessageId"] = properties.MessageId;
                    }
                    if (property.Headers != null && property.Headers.Count > 0)
                    {
                        properties.Headers = property.Headers;
                        arguments["Headers"] = properties.Headers;
                    }
                    if (!string.IsNullOrEmpty(property.Experation))
                    {
                        properties.Expiration = property.Experation;
                        arguments["Expiration"] = properties.Expiration;
                    }
                    if (property.DeliveryMode != null)
                    {
                        properties.DeliveryMode = property.DeliveryMode.Value;
                        arguments["DeliverMode"] = properties.DeliveryMode;
                    }
                    if (!string.IsNullOrEmpty(property.CorrelationId))
                    {
                        properties.CorrelationId = property.CorrelationId;
                        arguments["CorrelationId"] = properties.CorrelationId;
                    }
                    if (!string.IsNullOrEmpty(property.ContentType))
                    {
                        properties.ContentType = property.ContentType;
                        arguments["ContentType"] = properties.ContentType;
                    }
                    if (!string.IsNullOrEmpty(property.ContentEncoding))
                    {
                        properties.ContentEncoding = property.ContentEncoding;
                        arguments["ContentEncodingn"] = properties.ContentEncoding;
                    }
                    if (!string.IsNullOrEmpty(property.ClusterId))
                    {
                        properties.ClusterId = property.ClusterId;
                        arguments["ClusterId"] = properties.ClusterId;
                    }
                    if (!string.IsNullOrEmpty(property.AppId))
                    {
                        properties.AppId = property.AppId;
                        arguments["AppId"] = properties.AppId;
                    }
                    if (!string.IsNullOrEmpty(property.Type))
                    {
                        properties.Type = property.Type;
                        arguments["Type"] = properties.Type;
                    }
                }

                ChannelMetaObject metaObject = this.CreateChanel(mode, exchangeName, queueName, routingKey, arguments);
                if (this.connection.IsOpen)
                {
                    metaObject.Channel.BasicPublish(exchangeName, routingKey, properties, data);
                }
            }
            catch (Exception e)
            {
                this.Error?.Invoke(this, new ExceptionEventArgs() { Exception = e, Message = "发送消息时出现异常" }, true);
            }
        }
        #endregion
    }
}